
package com.example.demo;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Controller;

import io.debezium.config.Configuration;
import io.debezium.data.Envelope;
import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;

/**
 * Description: TODO <br/>
 * Date: 2021年9月16日 下午7:02:17 <br/>
 * 
 * @author wangbing
 */
@Controller
public class DbzmMysqlConnetor {

	@Autowired
	private JdbcTemplate jdbcTemplate;

	public void dbzConfig() throws IOException {

		// 使用MySQL连接器定义Debezium引擎的配置...
		final Properties props = new Properties();
		props.setProperty("name", "engine");// 连接器的唯一名称。尝试使用相同名称再次注册将失败。（所有 Kafka Connect 连接器都需要此属性。）
		props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");// 连接器的 Java类的名称。始终io.debezium​.connector.mysql.MySqlConnector为 MySQL 连接器使用值。
		props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");// 负责连接器偏移持久性的Java 类的名称。
		props.setProperty("offset.storage.file.filename", "D:/offsets.dat");// 要存储偏移量的文件路径
		props.setProperty("offset.flush.interval.ms", "60000");// 尝试提交偏移量的时间间隔。默认值为 1 分钟。
		/* 开始连接器属性 */
		props.setProperty("database.hostname", "localhost");// MySQL 数据库服务器的 IP 地址或主机名。
		props.setProperty("database.port", "3307");// MySQL 数据库服务器的端口号。
		props.setProperty("database.user", "root");// MYSQL 数据库服务器的账户
		props.setProperty("database.password", "123");// MYSQL 数据库服务器的密码
		props.setProperty("database.server.id", "8023");// 此数据库客户端的数字 ID，在 MySQL 集群中所有当前运行的数据库进程中必须是唯一的。
		props.setProperty("database.server.name", "my-app-connector");// 标识并为受监控的特定 MySQL 数据库服务器/集群提供命名空间的逻辑名称。逻辑名称在所有其他连接器中应该是唯一的，因为它用作从此连接器发出的所有Kafka 主题名称的前缀。只能使用字母数字字符和下划线。
		props.setProperty("database.include.list", "wtest");// 指定以逗号分隔的数据库列表
//		  props.setProperty("database.exclude.list", "wtest");//排除以逗号分隔的数据库列表
//		  props.setProperty("table.include.list", "test");//指定以逗号分隔的表名列表
//		  props.setProperty("column.include.list", "id");//指定以逗号分隔的列名列表
		props.setProperty("include.schema.changes", "false");// 指定连接器是否应将数据库结构的更改(DDL)发布到与数据库服务器
		props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");
		props.setProperty("database.history.file.filename", "D:/dbhistory.dat");
		Configuration config = Configuration.from(props);

		// 用配置创建引擎
		try (DebeziumEngine<RecordChangeEvent<SourceRecord>> engine = DebeziumEngine
				.create(ChangeEventFormat.of(Connect.class)).using(props).notifying(this::synData).build()) {
			Executors.newSingleThreadExecutor().execute(engine);

			// 异步运行引擎
			ExecutorService executor = Executors.newSingleThreadExecutor();
			executor.execute(engine);

			// Do something else or wait for a signal or an event
		}
	}

	public void synData(RecordChangeEvent<SourceRecord> record) {
		  System.out.println("record.record().value()==="+record.record().value());
		  if (record.record().value()!=null) {
			  //获取日志内容
			  Struct value = (Struct) record.record().value();
			  //获取操作类型
			  Field field = value.schema().field("op");
				
				  if (field != null) { 
					  Envelope.Operation operation =
				  Envelope.Operation.forCode((String) value.get("op")) ;
				  System.out.println("operation==="+operation); //增加,修改,删除操作数据 
				  if(operation  !=Envelope.Operation.READ ) { 
					  String sql = ""; 
				  if (operation ==  Envelope.Operation.CREATE || operation == Envelope.Operation.UPDATE) {//新增
				  //获取结构体内容 
					  Struct struct = (Struct) value.get("after"); Integer id = (Integer)
				  struct.get("id"); String name = (String) struct.get("name");
				  System.out.println("id===="+id+" , name====="+name); sql =
				  "insert into test (id,name) values ("+id+",'"+
				  name+"') on duplicate key update" + " id="+id+",name='"+name+"'";
				  jdbcTemplate.execute(sql);
				  
				  }else if (operation == Envelope.Operation.DELETE) {//删除 //获取结构体内容
					  Struct
				  struct = (Struct) value.get("before"); Integer id = (Integer)
				  struct.get("id"); String name = (String) struct.get("name");
				  System.out.println("id===="+id+" , name====="+name); sql =
				  "delete from test where id ="+id; jdbcTemplate.execute(sql);
				  
				  } } }
				 
		}
		  
	
	  }

}
